package io.github.quickmsg.core.protocol;

import cn.hutool.core.date.DatePattern;
import io.github.quickmsg.common.auth.AuthManager;
import io.github.quickmsg.common.auth.enums.ConnectReasonEnum;
import io.github.quickmsg.common.channel.MqttChannel;
import io.github.quickmsg.common.context.ContextHolder;
import io.github.quickmsg.common.context.ReceiveContext;
import io.github.quickmsg.common.enums.DisconnectStatus;
import io.github.quickmsg.common.integrate.Integrate;
import io.github.quickmsg.common.integrate.SubscribeTopic;
import io.github.quickmsg.common.integrate.channel.IntegrateChannels;
import io.github.quickmsg.common.integrate.topic.IntegrateTopics;
import io.github.quickmsg.common.log.LogEvent;
import io.github.quickmsg.common.log.LogManager;
import io.github.quickmsg.common.log.LogStatus;
import io.github.quickmsg.common.message.mqtt.ConnectMessage;
import io.github.quickmsg.common.metric.CounterType;
import io.github.quickmsg.common.protocol.Protocol;
import io.github.quickmsg.common.utils.JacksonUtil;
import io.github.quickmsg.common.utils.MqttMessageUtils;
import io.github.quickmsg.core.mqtt.MqttReceiveContext;
import io.github.quickmsg.common.handler.MyInboundIdleStateHandler;
import io.netty.handler.codec.mqtt.*;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.time.DateFormatUtils;
import reactor.util.context.ContextView;

import java.util.ArrayList;
import java.util.Date;
import java.util.Optional;
import java.util.Set;

/**
 * @author luxurong
 */
@Slf4j
public class ConnectProtocol implements Protocol<ConnectMessage> {

    private static final int MILLI_SECOND_PERIOD = 1_000;


    @Override
    public void parseProtocol(ConnectMessage connectMessage, MqttChannel mqttChannel, ContextView contextView) {
        ReceiveContext<?> receiveContext = contextView.get(ReceiveContext.class);
        LogManager logManager = receiveContext.getLogManager();
        MqttReceiveContext mqttReceiveContext = (MqttReceiveContext) contextView.get(ReceiveContext.class);
        String clientIdentifier = mqttChannel.getClientId();
        Integrate integrate = mqttReceiveContext.getIntegrate();
        IntegrateChannels channels = integrate.getChannels();
        IntegrateTopics<SubscribeTopic> topics = integrate.getTopics();
        AuthManager authManager = mqttReceiveContext.getAuthManager();
        /*protocol version support*/
        if (MqttVersion.MQTT_3_1_1 != connectMessage.getVersion() && MqttVersion.MQTT_3_1 != connectMessage.getVersion() && MqttVersion.MQTT_5 != connectMessage.getVersion()) {
            mqttChannel.write(MqttMessageUtils.buildConnectAck(MqttConnectReturnCode.CONNECTION_REFUSED_UNACCEPTABLE_PROTOCOL_VERSION, false));
            mqttChannel.close();
            return;
        }
        // 一个连接中发送多次connect报文
        if (channels.exists(mqttChannel.getClientId()) && channels.get(mqttChannel.getClientId()).getId().equals(mqttChannel.getId())) {
            mqttChannel.write(MqttMessageUtils.buildDisConnect(MqttReasonCodes.Disconnect.PROTOCOL_ERROR));
            mqttChannel.close();
            return;
        }
        String username = Optional.ofNullable(connectMessage.getAuth()).map(MqttChannel.Auth::getUsername).orElse(null);
        byte[] password = Optional.ofNullable(connectMessage.getAuth()).map(MqttChannel.Auth::getPassword).orElseGet(() -> new byte[]{});
        authManager.auth(username, password, clientIdentifier).subscribe(connectReasonEnum -> {
            if (ConnectReasonEnum.SUCCESS.equals(connectReasonEnum)) {
                receiveContext.getMetricManager().getMetricRegistry().getMetricCounter(CounterType.CONNECT).increment();
                receiveContext.getMetricManager().getMetricRegistry().getMetricCounter(CounterType.CONNECT_EVENT).increment();

                mqttChannel.setConnectCache(connectMessage.getCache(receiveContext.getIntegrate().getCluster().getLocalNode()));
                mqttChannel.getConnectCache().setWill(connectMessage.getWill());

                logManager.printInfo(mqttChannel, LogEvent.CONNECT, LogStatus.SUCCESS, "channelId:" + mqttChannel.getId() + JacksonUtil.bean2Json(connectMessage.getCache(receiveContext.getIntegrate().getCluster().getLocalNode())));

                mqttChannel.setAuthTime(DateFormatUtils.format(new Date(), DatePattern.NORM_DATETIME_PATTERN));

                /*registry unread event close channel */
                mqttChannel.getConnection().channel().pipeline().remove(MyInboundIdleStateHandler.class);
                long readIdle = (long) (connectMessage.getKeepalive() * MILLI_SECOND_PERIOD * 1.5);
                mqttChannel.getConnection().channel().pipeline()
                        .addFirst(mqttReceiveContext.getComputeLoopResources().onServer(true),
                                new MyInboundIdleStateHandler(readIdle, (s) -> this.logHeartClose(logManager, mqttChannel)));

                /* registry new channel*/
                channels.add(mqttChannel.getClientId(), mqttChannel);

                /* registry close mqtt channel event*/
                mqttChannel.registryClose(channel -> {
                    if (mqttChannel.getDisconnectStatus() == null) {
                        mqttChannel.setDisconnectStatus(DisconnectStatus.EX_DISCONNECT);
                    }
                    this.close(mqttChannel, mqttReceiveContext);
                });

                mqttChannel.write(MqttMessageUtils.buildConnectAck(MqttConnectReturnCode.CONNECTION_ACCEPTED, false));

            } else {
                logManager.printInfo(mqttChannel, LogEvent.CONNECT, LogStatus.FAILED, JacksonUtil.bean2Json(connectMessage.getCache(receiveContext.getIntegrate().getCluster().getLocalNode())));
                MqttConnAckMessage mqttConnAckMessage;
                if (ConnectReasonEnum.CLIENT_ID_NOT_VALID.equals(connectReasonEnum)) {
                    mqttConnAckMessage = MqttMessageUtils.buildConnectAck(MqttConnectReturnCode.CONNECTION_REFUSED_CLIENT_IDENTIFIER_NOT_VALID, false);
                } else {
                    mqttConnAckMessage = MqttMessageUtils.buildConnectAck(MqttConnectReturnCode.CONNECTION_REFUSED_BAD_USER_NAME_OR_PASSWORD, false);
                }
                mqttChannel.write(mqttConnAckMessage);

            }
        });


    }

    private void logHeartClose(LogManager logManager, MqttChannel mqttChannel) {
        mqttChannel.setDisconnectStatus(DisconnectStatus.HEART_BEAT_TIMEOUT);
        mqttChannel.write(MqttMessageUtils.buildDisConnect(MqttReasonCodes.Disconnect.KEEP_ALIVE_TIMEOUT));
        mqttChannel.close();
    }


    @Override
    public Class<ConnectMessage> getClassType() {
        return ConnectMessage.class;
    }


    private void close(MqttChannel mqttChannel, MqttReceiveContext mqttReceiveContext) {
        LogManager logManager = ContextHolder.getReceiveContext().getLogManager();
        LogEvent logEvent;
        if (DisconnectStatus.DISCONNECT.equals(mqttChannel.getDisconnectStatus())) {
            logEvent = LogEvent.DISCONNECT;
        } else if (DisconnectStatus.RECONNECT.equals(mqttChannel.getDisconnectStatus())) {
            logEvent = LogEvent.RECONNECT;
        } else if (DisconnectStatus.HEART_BEAT_TIMEOUT.equals(mqttChannel.getDisconnectStatus())) {
            logEvent = LogEvent.HEART_TIMEOUT;
        } else {
            logEvent = LogEvent.EX_DISCONNECT;
        }
        logManager.printInfo(mqttChannel, logEvent, LogStatus.SUCCESS, "channelId:" + mqttChannel.getId() + JacksonUtil.bean2Json(mqttChannel.getConnectCache()));

        ContextHolder.getReceiveContext().getMetricManager().getMetricRegistry().getMetricCounter(CounterType.CONNECT).decrement();
        ContextHolder.getReceiveContext().getMetricManager().getMetricRegistry().getMetricCounter(CounterType.DIS_CONNECT_EVENT).increment();
        // 移除channel
        final MqttChannel.Will willMessage = mqttChannel.getConnectCache().getWill();
        mqttReceiveContext.getIntegrate().getChannels().remove(mqttChannel);
        // 清除topics
        IntegrateTopics<SubscribeTopic> topics = mqttReceiveContext.getIntegrate().getTopics();
        topics.removeTopic(mqttChannel, new ArrayList<>(mqttChannel.getTopics()));
        // 清除重试程序
        mqttReceiveContext.getRetryManager().clearRetry(mqttChannel);
        // 发送断开消息
//        DisConnectMessage disConnectMessage = new DisConnectMessage(mqttChannel);
//        mqttReceiveContext.getIntegrate().getProtocolAdaptor().chooseProtocol(disConnectMessage);
        // 发送遗嘱消息
        if (willMessage != null && !DisconnectStatus.DISCONNECT.equals(mqttChannel.getDisconnectStatus())) {
            Set<SubscribeTopic> subscribeTopics = topics.getMqttChannelsByTopic(willMessage.getWillTopic());
            if (subscribeTopics != null) {
                subscribeTopics.forEach(subscribeTopic -> {
                    MqttChannel channel = subscribeTopic.getMqttChannel();
                    MqttQoS mqttQoS = subscribeTopic.minQos(willMessage.getMqttQoS());
                    channel.sendPublish(mqttQoS, willMessage.toPublishMessage());
                });
            }
        }

    }


}
